#include "ib.h"
#include "basic_tool.h"
#include "mpz_tool.h"
#include "memreuse.h"
#include "config.h"
#include "crypto_tool.h"
#include <errno.h>

IB::IB()
{
  pthread_mutex_init(&m_mutexSlidingWindow, NULL);
  pthread_cond_init (&m_cvSlidingWindow, NULL);
  m_listSlidingWindow.clear();
  m_bExit = false;
  m_pUdpAgent = NULL;
  m_id = 0;
  m_capacity = 0;
  m_buf = NULL;
  m_len = 0;
  m_subresid = 0;
  m_blockid = -1;
  m_totalblocks = 0;
  m_limit = -1;
}

IB::~IB()
{
  pthread_mutex_destroy(&m_mutexSlidingWindow);
  pthread_cond_destroy(&m_cvSlidingWindow);
  if (m_pUdpAgent != NULL)
	delete m_pUdpAgent;
}

void IB::LoadFile(TagReq * pTag)
{
  if (pTag->fileIndex != -1)
	{
	  stringstream ss;
	  ss<<Config::m_output_database_dir<<pTag->fileIndex/5000<<"/"<<pTag->fileIndex%5000;
	  if (BasicTool::Load_File(pTag->buf, pTag->len, ss.str()))
		{
		  pTag->fileIndex = -1;			// send end of response later
		}
	  pTag->totalblocks = ceil(pTag->len/(double)BLOCK_LEN);
	  if ((pTag->totalblocks - 1) * BLOCK_LEN == (int)pTag->len)
		{
		  pTag->totalblocks--;		// deal with precision error if dividable
		}
	}
}

bool IB::SendSingleRecordBlocks(TagReq * pTag)
{
  if (pTag == NULL)
	return false;				// waiting for new tags coming
  int index = pTag->fileIndex;
  if (m_buf == NULL && index != -1 && m_limit != 0) // if new file loadable
	{
// 	  stringstream ss;
// 	  ss<<Config::m_output_database_dir<<index/5000<<"/"<<index%5000;
// 	  if (BasicTool::Load_File(m_buf, m_len, ss.str()))
// 		{
// 		  index = -1;			// send end of response later
// 		}
// 	  m_totalblocks = ceil(m_len/(double)BLOCK_LEN);
// 	  if ((m_totalblocks - 1) * BLOCK_LEN == (int)m_len)
// 		{
// 		  m_totalblocks--;		// deal with precision error if dividable
// 		}
	  m_buf = pTag->buf;
	  m_len = pTag->len;
	  m_totalblocks = pTag->totalblocks;
	}
  bool bexit = false;			// all responses are sent;
  bool bOneRecordSent = false;	// one single response is sent
  while (!bexit && (!bOneRecordSent || m_limit == 0))
	{
	  UdpResponse * pResponse = new UdpResponse;
      pResponse->ID() = m_id++;
	  pResponse->Type() = 3;
	  pResponse->ResponseId() = m_reqid;
	  pResponse->SubResponseId() = m_subresid;
	  pResponse->BlockId() = m_blockid;
	  pResponse->TotalBlocks() = m_totalblocks;

	  if (index == -1 || m_limit == 0) // note not limit<=0. (limit=-1 means infinite)
		{
		  // send end of response
		  pResponse->TotalBlocks() = 0;
		  pTag->bDone = true;
		  bexit = true;
		}
	  else if (m_blockid == -1)	// send key
		{
		  pResponse->SetBlock(pTag->strEncK_Ki.c_str(), pTag->strEncK_Ki.length());
		  m_blockid ++;
		}
	  else
		{
		  // send packet
		  char * output = m_buf + m_blockid * BLOCK_LEN;
		  if (m_blockid == m_totalblocks - 1)
			{
			  pResponse->SetBlock(output, m_len - m_blockid * BLOCK_LEN);
			}
		  else
			{
			  pResponse->SetBlock(output, BLOCK_LEN);
			}
		  m_blockid ++;
		  if (m_blockid >= m_totalblocks)
			{
			  if (m_buf != NULL)
				{
				  delete[] m_buf;
				  m_buf = NULL;
				}
			  m_len = 0;
			  m_subresid++;
			  m_blockid = -1;
			  m_limit--;
			  bOneRecordSent = true;
			  pTag->bDone = true;
              if (m_subresid % 100 == 0)
                {
                  cout<<m_subresid<<" outputted"<<endl;
                }
			}
		}


	  pthread_mutex_lock(&m_mutexSlidingWindow);
	  m_pUdpAgent->Send((char *)pResponse, 
						sizeof(UdpResponse));
#ifdef _DEBUG
	  cout<<BasicTool::TimeStamp()<<" sending id="<<pResponse->ID()<<endl;
#endif
	  struct Response res;
	  res.pRes = pResponse;
	  BasicTool::TimeStamp(res.milli);
	  res.count = 1;
	  m_listSlidingWindow.push_back(res);
	  size_t wsize = m_listSlidingWindow.size();
	  pthread_mutex_unlock(&m_mutexSlidingWindow);

	  if (wsize >= (size_t) Config::m_sliding_window)
		{
		  break;
		}
	}
  return bexit;
}


int IB::OnNewRequest(UdpRequest * pRequest)
{
  list<struct Response>::iterator it;

  struct entry * et = NULL;
  m_buf = NULL;
  m_len = 0;
  m_reqid = pRequest->RequestId();
  m_subresid = 0;
  m_blockid = -1;
  m_totalblocks = 0;
  m_limit = pRequest->Limit();
  m_currentTagId = pRequest->TagId();
  TagReq * pTag = new TagReq;
  string strTag(pRequest->GetTag(), HASH_LEN);
  //  string strIndKey(pRequest->GetIndKey(), HASH_LEN);
  et = m_pHashTable->Lookup (NULL, strTag);
  if (et != NULL)
	{
	  pTag->fileIndex = et->index;
#ifdef _DEBUG
	  cout<<"fileindex "<<pTag->fileIndex<<endl;
#endif
	  pTag->tagid = pRequest->TagId();
	  pTag->strEncK_Ki = et->strEncK_Ki;
	  LoadFile(pTag);
	}
  else
	{
	  pTag->fileIndex = -1;
	  pTag->tagid = pRequest->TagId();
#ifdef _DEBUG
	  cout<<"record not found: "<<strTag<<endl;
#endif
	}
  m_dqTags.push_back(pTag);

  struct timespec to;

  while (1)
	{
	  pthread_mutex_lock(&m_mutexSlidingWindow);
	  size_t wsz = m_listSlidingWindow.size();
	  if (m_bExit)				// release any memory before exit
		{
		  while (m_dqTags.size() > 0)
			{
			  pTag = m_dqTags.front();
			  delete pTag;
			  m_dqTags.pop_front();
			}
		}
	  pthread_mutex_unlock(&m_mutexSlidingWindow);

	  if (wsz == 0 && m_bExit)
		break;

	  pthread_mutex_lock(&m_mutexSlidingWindow);
	  while (m_listSlidingWindow.size() >= (size_t) Config::m_sliding_window || 
			 (m_bExit && m_listSlidingWindow.size() > 0))
		{
		  clock_gettime(CLOCK_REALTIME, &to);
		  to.tv_sec += 1;
		  int rc = pthread_cond_timedwait(&m_cvSlidingWindow, &m_mutexSlidingWindow, &to);
		  // if timeout
          if (rc == ETIMEDOUT)
            {
			  it = m_listSlidingWindow.begin();
			  if (it->count <= 200) // quit after 200 retries
				{
				  m_pUdpAgent->Send((char *)(it->pRes), 
                                    sizeof(UdpResponse));
				  BasicTool::TimeStamp(it->milli);
				  it->count++;
				  cout<<BasicTool::TimeStamp()<<" receive timeout, resend "<<it->pRes->ID()<<endl;
				}
			  else
				{
				  cerr<<"connection end"<<endl;
				  while (!m_listSlidingWindow.empty())
					{
					  UdpResponse * pRes = (m_listSlidingWindow.front()).pRes;
					  m_listSlidingWindow.pop_front();
					  delete pRes;
					}
 
				  m_bExit = true;
				  break;		// quit if ten trials on the same packet
				}
            }		  // a new ack comes but window is full
		  else if (m_listSlidingWindow.size() >= (size_t) Config::m_sliding_window || 
			  (m_bExit && m_listSlidingWindow.size() > 0) )
			{

			  // if reaching here, client and server's window head is the same
			  it = m_listSlidingWindow.begin();
			  unsigned long milli;
			  BasicTool::TimeStamp(milli);
              for (int i = 0; i < 32 && it != m_listSlidingWindow.end(); i++)
                {
				  if ((1<<i) & m_capacity)
					{
					  ++it;
					  continue;
					}
				  // if last packet is sent within 100ms
				  if (milli - it->milli < 100 || it->count > 2) 
					{
					  ++it;
					  continue;					  
					}
                  UdpResponse * pRes = it->pRes;
                  m_pUdpAgent->Send((char *)pRes, 
                                    sizeof(UdpResponse));
				  BasicTool::TimeStamp(it->milli);
				  it->count++;

				  ++it;
#ifdef _DEBUG
                  cout<<BasicTool::TimeStamp()<<" resending id="<<pRes->ID()<<endl;
#endif
                }
			}
		}

	  // delete old tags;
	  if (m_dqTags.size() > 0)
		pTag = m_dqTags.front();	// null if m_dqtags.size()==0
	  else
		pTag = NULL;
	  while (pTag != NULL && pTag->bDone)
		{
		  delete pTag;
		  m_dqTags.pop_front();
		  if (m_dqTags.size() > 0)
			pTag = m_dqTags.front();	// null if m_dqtags.size()==0
		  else
			pTag = NULL;
		}
	  pthread_mutex_unlock(&m_mutexSlidingWindow);

	  if (!m_bExit && pTag != NULL)
		{
		  m_bExit = SendSingleRecordBlocks(pTag);
		}
	}
  cout<<"OnNewRequest exit"<<endl;
  return 0;
}

int IB::ReceiveAck ()
{
  UdpAck * pAck = new UdpAck;
  struct entry * et = NULL;
  TagReq * pTag = NULL;
  while (1)
	{
	  pthread_mutex_lock(&m_mutexSlidingWindow);
	  if (pTag != NULL)
		m_dqTags.push_back(pTag);
	  size_t wsz = m_listSlidingWindow.size();
	  pthread_mutex_unlock(&m_mutexSlidingWindow);
	  if (wsz == 0 && m_bExit)
		break;

      bool bRead = m_pUdpAgent->IsReadable(5000000);
      if (!bRead)
        continue;

      m_pUdpAgent->Recv((char*)pAck, sizeof(UdpAck));

	  pthread_mutex_lock(&m_mutexSlidingWindow);
#ifdef _DEBUG
	  cout<<BasicTool::TimeStamp()<<" receive ack id="<<pAck->ID()<<" capacity="<<pAck->Capacity()<<endl;
#endif
	  while (!m_listSlidingWindow.empty())
		{
		  UdpResponse * pRes = (m_listSlidingWindow.front()).pRes;
		  if (pRes->ID() >= pAck->ID())
			{
			  break;
			}
		  m_listSlidingWindow.pop_front();
		  delete pRes;
		}
      m_capacity = pAck->Capacity();
	  pthread_cond_signal(&m_cvSlidingWindow); // signal as long as an ACK arrives
	  pthread_mutex_unlock(&m_mutexSlidingWindow);

	  if (!m_bExit && pAck->TagId() > m_currentTagId) // do following only if new tag coming
		{
		  m_currentTagId = pAck->TagId();
		  pTag = new TagReq;
		  string strTag(pAck->GetTag(), HASH_LEN);
		  //		  string strIndKey(pAck->GetIndKey(), HASH_LEN);
		  et = m_pHashTable->Lookup (NULL, strTag);
		  if (et != NULL)
			{
			  pTag->fileIndex = et->index;
#ifdef _DEBUG
			  cout<<"fileindex "<<pTag->fileIndex<<endl;
#endif
			  pTag->tagid = pAck->TagId();
			  pTag->strEncK_Ki = et->strEncK_Ki;
			  LoadFile(pTag);
			}
		  else
			{
			  pTag->fileIndex = -1;
			  pTag->tagid = pAck->TagId();
#ifdef _DEBUG
			  cout<<"file not found"<<endl;
#endif
			}
		}
	  else
		{
		  pTag = NULL;
		}
	}
  delete pAck;
  cout<<"RecvAck exit"<<endl;
  return 0;
}

void IB::SetupUdp(unsigned short port, const struct sockaddr_in& sockaddClient)
{
  m_pUdpAgent = new UdpAgent;
  while (m_pUdpAgent->Bind(Config::m_ib_ip, port++) < 0)
    {
      cout<<"port "<<port-1<<" is in use"<<endl;
    }
  m_pUdpAgent->Connect(sockaddClient);
  m_pUdpAgent->SetNonBlocking();
}
